package kn.pulldata.task;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.lang.UUID;
import cn.hutool.crypto.digest.DigestUtil;
import cn.hutool.http.HttpRequest;
import cn.hutool.http.HttpResponse;
import cn.hutool.http.HttpUtil;
import cn.hutool.json.JSONUtil;
import com.google.common.hash.HashFunction;
import com.google.common.hash.Hashing;
import kn.pulldata.common.RabbitMqCon;
import kn.pulldata.common.RedisKey;
import kn.pulldata.req.MatchDistanceReq;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.time.LocalDate;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

/**
 *  拉取待匹配文本发送到消息队列中
 */
@Component
@Slf4j
public class PullMatchDataTask {

    @Value("${monitor.getMatchData.url}")
    private String getMatchDataUrl;


    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    @Qualifier("redisTemplateWithString")
    private RedisTemplate<String,String> redisTemplate;
    final int seed =100;
    private final ConcurrentHashMap<Long, Long> dataIdMap = new ConcurrentHashMap<>();
    private static final long EXPIRATION_TIME = 10 * 60 * 1000; // 10分钟
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);

    public PullMatchDataTask() {
        // 定时清理任务
        scheduler.scheduleAtFixedRate(this::cleanupExpiredDataIds, 10, 10, TimeUnit.MINUTES);
    }

    private void cleanupExpiredDataIds() {
        long currentTime = System.currentTimeMillis();
        dataIdMap.entrySet().removeIf(entry -> currentTime - entry.getValue() > EXPIRATION_TIME);
        log.info("清理过期的dataID，当前size: {}", dataIdMap.size());
    }
    private final ConcurrentHashMap<LocalDate, AtomicLong> dateCount = new ConcurrentHashMap<>();

    @Scheduled(fixedRate = 200) // 每隔200毫秒拉取待匹配数据
    @Async
    public void PullMatchDataTask61() {
        LocalDate currentDate = LocalDate.now();
        // 如果当前日期没有对应的计数器，则创建一个新的 AtomicLong 并初始化为 0
        dateCount.putIfAbsent(currentDate, new AtomicLong(0));
        AtomicLong atomicLongForToday = dateCount.get(currentDate);
        boolean stop = true;
        while (stop){
            log.info("PullMatchDataTask61开始执行");
            String format = String.format(getMatchDataUrl, 61);
            log.info("请求地址：{}", format);
            HttpRequest httpRequest = HttpUtil.createGet(format, true);
            HttpResponse httpResponse = httpRequest.execute();
            if (httpResponse != null && httpResponse.body() != null) {
                String body = httpResponse.body();
                List<MatchDistanceReq> matchDistanceReqList = JSONUtil.parseArray(body).toList(MatchDistanceReq.class);
                if (CollUtil.isNotEmpty(matchDistanceReqList)) {
                    List<MatchDistanceReq> newMatchDistanceReqList = new ArrayList<>();
                    for (MatchDistanceReq req : matchDistanceReqList) {
                        Long dataId = req.getDataID();
                        HashFunction hashFunction128 = Hashing.murmur3_128(seed);
                        Long uniqueKeyHash = hashFunction128.hashUnencodedChars(req.getSource() + "|" + req.getSourceUrl()).asLong();
                        if (dataIdMap.putIfAbsent(uniqueKeyHash, System.currentTimeMillis()) != null) {
                            log.info("重复的dataID: {}", dataId);
                        } else {
                            newMatchDistanceReqList.add(req);
                            atomicLongForToday.incrementAndGet();
                        }
                    }

                    log.info("最新时间{}的总量是{}", currentDate, atomicLongForToday.get());
                    // 新数据，准备发送
                    String messageId = UUID.randomUUID().toString();
                    Message message = MessageBuilder.withBody(JSONUtil.toJsonStr(newMatchDistanceReqList).getBytes(StandardCharsets.UTF_8))
                            .setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)
                            .setMessageId(messageId)
                            .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
                            .build();
                    CorrelationData correlationData = new CorrelationData(messageId);
                    rabbitTemplate.send(RabbitMqCon.TOPIC_EXCHANGE, RabbitMqCon.KEY, message, correlationData);
                } else {
                    log.info("61拉取数据为空");
                    stop = false;
                }
            }
        }

        dateCount.remove(LocalDate.now().minusDays(1L));
        log.info("PullMatchDataTask61执行结束");
    }

    @Scheduled(fixedRate = 200) // 每隔200毫秒拉取待匹配数据
    @Async
    public void PullMatchDataTask62() {
        LocalDate currentDate = LocalDate.now();
        // 如果当前日期没有对应的计数器，则创建一个新的 AtomicLong 并初始化为 0
        dateCount.putIfAbsent(currentDate, new AtomicLong(0));
        AtomicLong atomicLongForToday = dateCount.get(currentDate);
        boolean stop = true;
        while (stop){
            log.info("PullMatchDataTask62开始执行");
            String format = String.format(getMatchDataUrl, 62);
            log.info("请求地址：{}", format);
            HttpRequest httpRequest = HttpUtil.createGet(format, true);
            HttpResponse httpResponse = httpRequest.execute();
            if (httpResponse != null && httpResponse.body() != null) {
                String body = httpResponse.body();
                List<MatchDistanceReq> matchDistanceReqList = JSONUtil.parseArray(body).toList(MatchDistanceReq.class);
                if (CollUtil.isNotEmpty(matchDistanceReqList)) {
                    List<MatchDistanceReq> newMatchDistanceReqList = new ArrayList<>();
                    for (MatchDistanceReq req : matchDistanceReqList) {
                        Long dataId = req.getDataID();
                        HashFunction hashFunction128 = Hashing.murmur3_128(seed);
                        Long uniqueKeyHash = hashFunction128.hashUnencodedChars(req.getSource() + "|" + req.getSourceUrl()).asLong();
                        if (dataIdMap.putIfAbsent(uniqueKeyHash, System.currentTimeMillis()) != null) {
                            log.info("重复的dataID: {}", dataId);
                        } else {
                            newMatchDistanceReqList.add(req);
                            atomicLongForToday.incrementAndGet();
                        }
                    }

                    log.info("最新时间{}的总量是{}", currentDate, atomicLongForToday.get());
                    // 新数据，准备发送
                    String messageId = UUID.randomUUID().toString();
                    Message message = MessageBuilder.withBody(JSONUtil.toJsonStr(newMatchDistanceReqList).getBytes(StandardCharsets.UTF_8))
                            .setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)
                            .setMessageId(messageId)
                            .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
                            .build();
                    CorrelationData correlationData = new CorrelationData(messageId);
                    rabbitTemplate.send(RabbitMqCon.TOPIC_EXCHANGE, RabbitMqCon.KEY, message, correlationData);
                } else {
                    log.info("62拉取数据为空");
                    stop = false;
                }
            }
        }

        
        log.info("PullMatchDataTask62执行结束");
    }

    @Scheduled(fixedRate = 200) // 每隔200毫秒拉取待匹配数据
    @Async
    public void PullMatchDataTask63() {
        LocalDate currentDate = LocalDate.now();
        // 如果当前日期没有对应的计数器，则创建一个新的 AtomicLong 并初始化为 0
        dateCount.putIfAbsent(currentDate, new AtomicLong(0));
        AtomicLong atomicLongForToday = dateCount.get(currentDate);
        boolean stop = true;
        while (stop){
            log.info("PullMatchDataTask63开始执行");
            String format = String.format(getMatchDataUrl, 63);
            log.info("请求地址：{}", format);
            HttpRequest httpRequest = HttpUtil.createGet(format, true);
            HttpResponse httpResponse = httpRequest.execute();
            if (httpResponse != null && httpResponse.body() != null) {
                String body = httpResponse.body();
                List<MatchDistanceReq> matchDistanceReqList = JSONUtil.parseArray(body).toList(MatchDistanceReq.class);
                if (CollUtil.isNotEmpty(matchDistanceReqList)) {
                    List<MatchDistanceReq> newMatchDistanceReqList = new ArrayList<>();
                    for (MatchDistanceReq req : matchDistanceReqList) {
                        Long dataId = req.getDataID();
                        HashFunction hashFunction128 = Hashing.murmur3_128(seed);
                        Long uniqueKeyHash = hashFunction128.hashUnencodedChars(req.getSource() + "|" + req.getSourceUrl()).asLong();
                        if (dataIdMap.putIfAbsent(uniqueKeyHash, System.currentTimeMillis()) != null) {
                            log.info("重复的dataID: {}", dataId);
                        } else {
                            newMatchDistanceReqList.add(req);
                            atomicLongForToday.incrementAndGet();
                        }
                    }

                    log.info("最新时间{}的总量是{}", currentDate, atomicLongForToday.get());
                    // 新数据，准备发送
                    String messageId = UUID.randomUUID().toString();
                    Message message = MessageBuilder.withBody(JSONUtil.toJsonStr(newMatchDistanceReqList).getBytes(StandardCharsets.UTF_8))
                            .setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)
                            .setMessageId(messageId)
                            .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
                            .build();
                    CorrelationData correlationData = new CorrelationData(messageId);
                    rabbitTemplate.send(RabbitMqCon.TOPIC_EXCHANGE, RabbitMqCon.KEY, message, correlationData);
                } else {
                    log.info("63拉取数据为空");
                    stop = false;
                }
            }
        }
        log.info("PullMatchDataTask63执行结束");
    }

    @Scheduled(fixedRate = 200) // 每隔200毫秒拉取待匹配数据
    @Async
    public void PullMatchDataTask64() {
        LocalDate currentDate = LocalDate.now();
        // 如果当前日期没有对应的计数器，则创建一个新的 AtomicLong 并初始化为 0
        dateCount.putIfAbsent(currentDate, new AtomicLong(0));
        AtomicLong atomicLongForToday = dateCount.get(currentDate);
        boolean stop = true;
        while (stop){
            log.info("PullMatchDataTask64开始执行");
            String format = String.format(getMatchDataUrl, 64);
            log.info("请求地址：{}", format);
            HttpRequest httpRequest = HttpUtil.createGet(format, true);
            HttpResponse httpResponse = httpRequest.execute();
            if (httpResponse != null && httpResponse.body() != null) {
                String body = httpResponse.body();
                List<MatchDistanceReq> matchDistanceReqList = JSONUtil.parseArray(body).toList(MatchDistanceReq.class);
                if (CollUtil.isNotEmpty(matchDistanceReqList)) {
                    List<MatchDistanceReq> newMatchDistanceReqList = new ArrayList<>();
                    for (MatchDistanceReq req : matchDistanceReqList) {
                        Long dataId = req.getDataID();
                        HashFunction hashFunction128 = Hashing.murmur3_128(seed);
                        Long uniqueKeyHash = hashFunction128.hashUnencodedChars(req.getSource() + "|" + req.getSourceUrl()).asLong();
                        if (dataIdMap.putIfAbsent(uniqueKeyHash, System.currentTimeMillis()) != null) {
                            log.info("重复的dataID: {}", dataId);
                        } else {
                            newMatchDistanceReqList.add(req);
                            atomicLongForToday.incrementAndGet();
                        }
                    }

                    log.info("最新时间{}的总量是{}", currentDate, atomicLongForToday.get());
                    // 新数据，准备发送
                    String messageId = UUID.randomUUID().toString();
                    Message message = MessageBuilder.withBody(JSONUtil.toJsonStr(newMatchDistanceReqList).getBytes(StandardCharsets.UTF_8))
                            .setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)
                            .setMessageId(messageId)
                            .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
                            .build();
                    CorrelationData correlationData = new CorrelationData(messageId);
                    rabbitTemplate.send(RabbitMqCon.TOPIC_EXCHANGE, RabbitMqCon.KEY, message, correlationData);
                } else {
                    log.info("64拉取数据为空");
                    stop = false;
                }
            }
        }

        
        log.info("PullMatchDataTask64执行结束");
    }

    @Scheduled(fixedRate = 200) // 每隔200毫秒拉取待匹配数据
    @Async
    public void PullMatchDataTask65() {
        LocalDate currentDate = LocalDate.now();
        // 如果当前日期没有对应的计数器，则创建一个新的 AtomicLong 并初始化为 0
        dateCount.putIfAbsent(currentDate, new AtomicLong(0));
        AtomicLong atomicLongForToday = dateCount.get(currentDate);
        boolean stop = true;
        while (stop){
            log.info("PullMatchDataTask65开始执行");
            String format = String.format(getMatchDataUrl, 65);
            log.info("请求地址：{}", format);
            HttpRequest httpRequest = HttpUtil.createGet(format, true);
            HttpResponse httpResponse = httpRequest.execute();
            if (httpResponse != null && httpResponse.body() != null) {
                String body = httpResponse.body();
                List<MatchDistanceReq> matchDistanceReqList = JSONUtil.parseArray(body).toList(MatchDistanceReq.class);
                if (CollUtil.isNotEmpty(matchDistanceReqList)) {
                    List<MatchDistanceReq> newMatchDistanceReqList = new ArrayList<>();
                    for (MatchDistanceReq req : matchDistanceReqList) {
                        Long dataId = req.getDataID();
                        HashFunction hashFunction128 = Hashing.murmur3_128(seed);
                        Long uniqueKeyHash = hashFunction128.hashUnencodedChars(req.getSource() + "|" + req.getSourceUrl()).asLong();
                        if (dataIdMap.putIfAbsent(uniqueKeyHash, System.currentTimeMillis()) != null) {
                            log.info("重复的dataID: {}", dataId);
                        } else {
                            newMatchDistanceReqList.add(req);
                            atomicLongForToday.incrementAndGet();
                        }
                    }

                    log.info("最新时间{}的总量是{}", currentDate, atomicLongForToday.get());
                    // 新数据，准备发送
                    String messageId = UUID.randomUUID().toString();
                    Message message = MessageBuilder.withBody(JSONUtil.toJsonStr(newMatchDistanceReqList).getBytes(StandardCharsets.UTF_8))
                            .setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)
                            .setMessageId(messageId)
                            .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
                            .build();
                    CorrelationData correlationData = new CorrelationData(messageId);
                    rabbitTemplate.send(RabbitMqCon.TOPIC_EXCHANGE, RabbitMqCon.KEY, message, correlationData);
                } else {
                    log.info("65拉取数据为空");
                    stop = false;
                }
            }
        }

        
        log.info("PullMatchDataTask65执行结束");
    }


    @Scheduled(fixedRate = 200) // 每隔200毫秒拉取待匹配数据
    @Async
    public void PullMatchDataTask66() {
        LocalDate currentDate = LocalDate.now();
        // 如果当前日期没有对应的计数器，则创建一个新的 AtomicLong 并初始化为 0
        dateCount.putIfAbsent(currentDate, new AtomicLong(0));
        AtomicLong atomicLongForToday = dateCount.get(currentDate);
        log.info("PullMatchDataTask66开始执行");
        boolean stop = true;
        while (stop){
            String format = String.format(getMatchDataUrl, 66);
            log.info("请求地址：{}", format);
            HttpRequest httpRequest = HttpUtil.createGet(format, true);
            HttpResponse httpResponse = httpRequest.execute();
            if (httpResponse != null && httpResponse.body() != null) {
                String body = httpResponse.body();
                List<MatchDistanceReq> matchDistanceReqList = JSONUtil.parseArray(body).toList(MatchDistanceReq.class);
                if (CollUtil.isNotEmpty(matchDistanceReqList)) {
                    List<MatchDistanceReq> newMatchDistanceReqList = new ArrayList<>();
                    for (MatchDistanceReq req : matchDistanceReqList) {
                        Long dataId = req.getDataID();
                        HashFunction hashFunction128 = Hashing.murmur3_128(seed);
                        Long uniqueKeyHash = hashFunction128.hashUnencodedChars(req.getSource() + "|" + req.getSourceUrl()).asLong();
                        if (dataIdMap.putIfAbsent(uniqueKeyHash, System.currentTimeMillis()) != null) {
                            log.info("重复的dataID: {}", dataId);
                        } else {
                            newMatchDistanceReqList.add(req);
                            atomicLongForToday.incrementAndGet();
                        }
                    }

                    log.info("最新时间{}的总量是{}", currentDate, atomicLongForToday.get());
                    // 新数据，准备发送
                    String messageId = UUID.randomUUID().toString();
                    Message message = MessageBuilder.withBody(JSONUtil.toJsonStr(newMatchDistanceReqList).getBytes(StandardCharsets.UTF_8))
                            .setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)
                            .setMessageId(messageId)
                            .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
                            .build();
                    CorrelationData correlationData = new CorrelationData(messageId);
                    rabbitTemplate.send(RabbitMqCon.TOPIC_EXCHANGE, RabbitMqCon.KEY, message, correlationData);
                } else {
                    log.info("66拉取数据为空");
                    stop = false;
                }
            }
        }

        
        log.info("PullMatchDataTask66执行结束");
    }

    @Scheduled(fixedRate = 200) // 每隔200毫秒拉取待匹配数据
    @Async
    public void PullMatchDataTask67() {
        LocalDate currentDate = LocalDate.now();
        // 如果当前日期没有对应的计数器，则创建一个新的 AtomicLong 并初始化为 0
        dateCount.putIfAbsent(currentDate, new AtomicLong(0));
        AtomicLong atomicLongForToday = dateCount.get(currentDate);
        log.info("PullMatchDataTask67开始执行");
        boolean stop = true;
        while (stop){
            String format = String.format(getMatchDataUrl, 67);
            log.info("请求地址：{}", format);
            HttpRequest httpRequest = HttpUtil.createGet(format, true);
            HttpResponse httpResponse = httpRequest.execute();
            if (httpResponse != null && httpResponse.body() != null) {
                String body = httpResponse.body();
                List<MatchDistanceReq> matchDistanceReqList = JSONUtil.parseArray(body).toList(MatchDistanceReq.class);
                if (CollUtil.isNotEmpty(matchDistanceReqList)) {
                    List<MatchDistanceReq> newMatchDistanceReqList = new ArrayList<>();
                    for (MatchDistanceReq req : matchDistanceReqList) {
                        Long dataId = req.getDataID();
                        HashFunction hashFunction128 = Hashing.murmur3_128(seed);
                        Long uniqueKeyHash = hashFunction128.hashUnencodedChars(req.getSource() + "|" + req.getSourceUrl()).asLong();
                        if (dataIdMap.putIfAbsent(uniqueKeyHash, System.currentTimeMillis()) != null) {
                            log.info("重复的dataID: {}", dataId);
                        } else {
                            newMatchDistanceReqList.add(req);
                            atomicLongForToday.incrementAndGet();
                        }
                    }

                    log.info("最新时间{}的总量是{}", currentDate, atomicLongForToday.get());
                    // 新数据，准备发送
                    String messageId = UUID.randomUUID().toString();
                    Message message = MessageBuilder.withBody(JSONUtil.toJsonStr(newMatchDistanceReqList).getBytes(StandardCharsets.UTF_8))
                            .setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)
                            .setMessageId(messageId)
                            .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
                            .build();
                    CorrelationData correlationData = new CorrelationData(messageId);
                    rabbitTemplate.send(RabbitMqCon.TOPIC_EXCHANGE, RabbitMqCon.KEY, message, correlationData);
                } else {
                    log.info("67拉取数据为空");
                    stop = false;
                }
            }
        }

        
        log.info("PullMatchDataTask67执行结束");
    }

    @Scheduled(fixedRate = 200) // 每隔200毫秒拉取待匹配数据
    @Async
    public void PullMatchDataTask68() {
        LocalDate currentDate = LocalDate.now();
        // 如果当前日期没有对应的计数器，则创建一个新的 AtomicLong 并初始化为 0
        dateCount.putIfAbsent(currentDate, new AtomicLong(0));
        AtomicLong atomicLongForToday = dateCount.get(currentDate);
        boolean stop = true;
        while (stop){
            log.info("PullMatchDataTask68开始执行");
            String format = String.format(getMatchDataUrl, 68);
            log.info("请求地址：{}", format);
            HttpRequest httpRequest = HttpUtil.createGet(format, true);
            HttpResponse httpResponse = httpRequest.execute();
            if (httpResponse != null && httpResponse.body() != null) {
                String body = httpResponse.body();
                List<MatchDistanceReq> matchDistanceReqList = JSONUtil.parseArray(body).toList(MatchDistanceReq.class);
                if (CollUtil.isNotEmpty(matchDistanceReqList)) {
                    List<MatchDistanceReq> newMatchDistanceReqList = new ArrayList<>();
                    for (MatchDistanceReq req : matchDistanceReqList) {
                        Long dataId = req.getDataID();
                        HashFunction hashFunction128 = Hashing.murmur3_128(seed);
                        Long uniqueKeyHash = hashFunction128.hashUnencodedChars(req.getSource() + "|" + req.getSourceUrl()).asLong();
                        if (dataIdMap.putIfAbsent(uniqueKeyHash, System.currentTimeMillis()) != null) {
                            log.info("重复的dataID: {}", dataId);
                        } else {
                            newMatchDistanceReqList.add(req);
                            atomicLongForToday.incrementAndGet();
                        }
                    }

                    log.info("最新时间{}的总量是{}", currentDate, atomicLongForToday.get());
                    // 新数据，准备发送
                    String messageId = UUID.randomUUID().toString();
                    Message message = MessageBuilder.withBody(JSONUtil.toJsonStr(newMatchDistanceReqList).getBytes(StandardCharsets.UTF_8))
                            .setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)
                            .setMessageId(messageId)
                            .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
                            .build();
                    CorrelationData correlationData = new CorrelationData(messageId);
                    rabbitTemplate.send(RabbitMqCon.TOPIC_EXCHANGE, RabbitMqCon.KEY, message, correlationData);
                } else {
                    log.info("68拉取数据为空");
                    stop = false;
                }
            }
        }

        
        log.info("PullMatchDataTask68执行结束");
    }
    @Scheduled(fixedRate = 200) // 每隔200毫秒拉取待匹配数据
    @Async
    public void PullMatchDataTask69() {
        LocalDate currentDate = LocalDate.now();
        // 如果当前日期没有对应的计数器，则创建一个新的 AtomicLong 并初始化为 0
        dateCount.putIfAbsent(currentDate, new AtomicLong(0));
        AtomicLong atomicLongForToday = dateCount.get(currentDate);
        boolean stop = true;
        while (stop){
            log.info("PullMatchDataTask69开始执行");
            String format = String.format(getMatchDataUrl, 69);
            log.info("请求地址：{}", format);
            HttpRequest httpRequest = HttpUtil.createGet(format, true);
            HttpResponse httpResponse = httpRequest.execute();
            if (httpResponse != null && httpResponse.body() != null) {
                String body = httpResponse.body();
                List<MatchDistanceReq> matchDistanceReqList = JSONUtil.parseArray(body).toList(MatchDistanceReq.class);
                if (CollUtil.isNotEmpty(matchDistanceReqList)) {
                    List<MatchDistanceReq> newMatchDistanceReqList = new ArrayList<>();
                    for (MatchDistanceReq req : matchDistanceReqList) {
                        Long dataId = req.getDataID();
                        HashFunction hashFunction128 = Hashing.murmur3_128(seed);
                        Long uniqueKeyHash = hashFunction128.hashUnencodedChars(req.getSource() + "|" + req.getSourceUrl()).asLong();
                        if (dataIdMap.putIfAbsent(uniqueKeyHash, System.currentTimeMillis()) != null) {
                            log.info("重复的dataID: {}", dataId);
                        } else {
                            newMatchDistanceReqList.add(req);
                            atomicLongForToday.incrementAndGet();
                        }
                    }

                    log.info("最新时间{}的总量是{}", currentDate, atomicLongForToday.get());
                    // 新数据，准备发送
                    String messageId = UUID.randomUUID().toString();
                    Message message = MessageBuilder.withBody(JSONUtil.toJsonStr(newMatchDistanceReqList).getBytes(StandardCharsets.UTF_8))
                            .setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)
                            .setMessageId(messageId)
                            .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
                            .build();
                    CorrelationData correlationData = new CorrelationData(messageId);
                    rabbitTemplate.send(RabbitMqCon.TOPIC_EXCHANGE, RabbitMqCon.KEY, message, correlationData);
                } else {
                    log.info("69拉取数据为空");
                    stop = false;
                }
            }

        }
        
        log.info("PullMatchDataTask69执行结束");
    }

    @Scheduled(fixedRate = 200) // 每隔200毫秒拉取待匹配数据
    @Async
    public void PullMatchDataTask70() {
        LocalDate currentDate = LocalDate.now();
        // 如果当前日期没有对应的计数器，则创建一个新的 AtomicLong 并初始化为 0
        dateCount.putIfAbsent(currentDate, new AtomicLong(0));
        AtomicLong atomicLongForToday = dateCount.get(currentDate);
        boolean stop = true;
        while (stop){
            log.info("PullMatchDataTask70开始执行");
            String format = String.format(getMatchDataUrl, 70);
            log.info("请求地址：{}", format);
            HttpRequest httpRequest = HttpUtil.createGet(format, true);
            HttpResponse httpResponse = httpRequest.execute();
            if (httpResponse != null && httpResponse.body() != null) {
                String body = httpResponse.body();
                List<MatchDistanceReq> matchDistanceReqList = JSONUtil.parseArray(body).toList(MatchDistanceReq.class);
                if (CollUtil.isNotEmpty(matchDistanceReqList)) {
                    List<MatchDistanceReq> newMatchDistanceReqList = new ArrayList<>();
                    for (MatchDistanceReq req : matchDistanceReqList) {
                        Long dataId = req.getDataID();
                        HashFunction hashFunction128 = Hashing.murmur3_128(seed);
                        Long uniqueKeyHash = hashFunction128.hashUnencodedChars(req.getSource() + "|" + req.getSourceUrl()).asLong();
                        if (dataIdMap.putIfAbsent(uniqueKeyHash, System.currentTimeMillis()) != null) {
                            log.info("重复的dataID: {}", dataId);
                        } else {
                            newMatchDistanceReqList.add(req);
                            atomicLongForToday.incrementAndGet();
                        }
                    }

                    log.info("最新时间{}的总量是{}", currentDate, atomicLongForToday.get());
                    // 新数据，准备发送
                    String messageId = UUID.randomUUID().toString();
                    Message message = MessageBuilder.withBody(JSONUtil.toJsonStr(newMatchDistanceReqList).getBytes(StandardCharsets.UTF_8))
                            .setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)
                            .setMessageId(messageId)
                            .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
                            .build();
                    CorrelationData correlationData = new CorrelationData(messageId);
                    rabbitTemplate.send(RabbitMqCon.TOPIC_EXCHANGE, RabbitMqCon.KEY, message, correlationData);
                } else {
                    log.info("70拉取数据为空");
                    stop = false;
                }
        }

        }
        
        log.info("PullMatchDataTask70执行结束");
    }
    private void saveRedis(List<MatchDistanceReq> redisList, String key) {
        // 构建批量操作的 Pipeline
        redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
            // Redis 的键名
            byte[] redisKey =key.getBytes(StandardCharsets.UTF_8);

            for (MatchDistanceReq matchDistanceReq : redisList) {
                if (matchDistanceReq != null) {
                    // 构建 uniqueKey
                    String uniqueKey = matchDistanceReq.getSourceUrl() + "|" + matchDistanceReq.getSourceSite();
                    // 对 uniqueKey 进行 MD5 加密
                    String uniqueKeyHash = DigestUtil.md5Hex16(uniqueKey);
                    // 将批量 SADD 操作加入 pipeline
                    connection.sAdd(redisKey, uniqueKeyHash.getBytes(StandardCharsets.UTF_8));
                }
            }
            return null; // 执行完 pipeline 后返回 null
        });
        // 设置过期时间
        redisTemplate.expire(key, 1, TimeUnit.HOURS);
    }
    public static void main(String[] args) {
//        NumberPoller numberPoller = new NumberPoller();
//
//        for (int i = numberPoller.currentNumber; i <= numberPoller.end; i++) {
//            System.out.println(i);
//        }
        // todo 优化用这种方式做
        ConcurrentHashMap<Long, Long> dataIdMap = new ConcurrentHashMap<>();

        List<String> list = new ArrayList<>();
        list.add("12344");
        list.add("T7867");
        list.add("78567");
        list.add("12344");
        int seed =100;
        HashFunction hashFunction128 = Hashing.murmur3_128(seed);
        for (String s : list) {
            long hash128Low = hashFunction128.hashUnencodedChars(s).asLong();
            if (dataIdMap.putIfAbsent(hash128Low, System.currentTimeMillis()) != null) {
                log.info("重复的dataID");
            } else {
                log.info("发送成功");
            }
        }

    }

    @Scheduled(cron = "0 0 3 */2 * ?")
    public void clearRedisKeys() {
        // 清除Redis的Key
        Boolean delete = redisTemplate.delete(RedisKey.DIRECTION_REPEAT_REDIS_KEY);
        log.info("执行清除Redis Key的任务{}={}" ,new Date(),delete);
    }

}
